package com.zhaotianhao.check.comboserver.task;

import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Set;

/**
 * @Author Z-7
 * @Date 2022/4/2
 */
@Slf4j
@Component
public class ComboTask {
    @Resource
    private KafkaTemplate kafkaTemplate;
    @Resource
    private RedisTemplate redisTemplate;

    //@Scheduled(cron = "0 * * * * ?")
    public void messageListen() {
        log.info("定时器执行");
        Set keys = redisTemplate.keys("message:*");
        if (CollectionUtil.isEmpty(keys)) {
            log.info("暂无未消费的信息");
            return;
        }
        // 有消息未消费
        for (Object key : keys) {
            // 取出
            Object o = redisTemplate.opsForValue().get(key);
            JSONObject jsonObject = JSONObject.parseObject(o.toString());
            if (System.currentTimeMillis() - jsonObject.getLong("createTime") > 10000) {
                // 如果消息超过10秒还没有消费  认为 消息丢失
                System.out.println((key + "").substring(8));
                // 消息补偿
                kafkaTemplate.send("combo", (key + "").substring(8) ,jsonObject.getString("message"));
            }
        }
    }
}
